#include "counter_service.h"

#include "redis_executor/scoped_reply.h"

using namespace std;

namespace jztzp {

class ReloadCountTask : public concurrency::Runnable {
  GenericCounterHandler * handler_;
  bool instant_load_;
 public:
  ReloadCountTask(GenericCounterHandler * handler, bool load_now) : handler_(handler), instant_load_(load_now) {
  }
  virtual void run() {
    LOG_INFO("start loading redis.");
    if (instant_load_ || time(NULL) % (24 * 60 * 60) <= 60 * 60) {
      if (handler_->ReloadRedis()) {
        LOG_INFO("loading redis SUCCESS.");
      } else {
        LOG_ERROR("loading redis ERROR.");
      }
    }
    handler_->timer_manager_.add(shared_ptr<concurrency::Runnable>(new ReloadCountTask(handler_, false)), 1000 * 60 * 60);
    LOG_INFO("restart load timer.");
  }
};

class UpdateCountTask : public concurrency::Runnable {
  GenericCounterHandler * handler_;
  int business_type_;
  int64_t business_id_;
  int delta_;
 public:
  UpdateCountTask(GenericCounterHandler * handler, int business_type, int64_t business_id, int delta) : handler_(handler), 
    business_type_(business_type), 
    business_id_(business_id), 
    delta_(delta) {
  }
  virtual void run() {
    if (handler_->UpdateDatabase(business_type_, business_id_, delta_)) {
      LOG_INFO("update count SUCCESS.");
    } else {
      LOG_ERROR("update count ERROR.");
    }
  }
};

GenericCounterHandler::GenericCounterHandler() :
    db_pool_("xingtan", "127.0.0.1:3306", "root", "") {

  thread_manager_ = concurrency::ThreadManager::newSimpleThreadManager(10, 20);
  shared_ptr<concurrency::PlatformThreadFactory> factory = shared_ptr<concurrency::PlatformThreadFactory>(new concurrency::PlatformThreadFactory());

  thread_manager_->threadFactory(factory);
  thread_manager_->start();

  timer_manager_.threadFactory(factory);
  timer_manager_.start();
  timer_manager_.add(shared_ptr<concurrency::Runnable>(new ReloadCountTask(this, true)), 1000);
}

GenericCounterHandler::~GenericCounterHandler() {
  thread_manager_->stop();
}

bool GenericCounterHandler::UpdateDatabase(int business_type, int64_t business_id, int delta) {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return false;
  }

  mysqlpp::Query query(cp->query());

  query << "INSERT INTO generic_counter(business_type, business_id, counter) VALUES (" 
        << business_type << ',' << business_id << ',' << delta << ") ON DUPLICATE KEY UPDATE counter=" 
        << "counter + " << delta;

  bool res = false;
  try {
    res = query.exec();
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR("update counter " << business_id << " error:" << e.what());
  }
  LOG_DEBUG("update counter comment " << business_id << " success=" << res);
  return res;
}

bool GenericCounterHandler::ReloadRedis() {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return false;
  }

  mysqlpp::Query query(cp->query());

  query << "SELECT business_type, business_id, counter FROM generic_counter";
  LOG_INFO("ReloadRedis() " << query.str());

  try{
    mysqlpp::UseQueryResult res = query.use();
    if (res) {
      const int MAX_BUF_LEN = 64;
      char cmd_buf[MAX_BUF_LEN];
      cmd_buf[MAX_BUF_LEN - 1] = '\0';

      while(mysqlpp::Row row = res.fetch_row()) {
        snprintf(cmd_buf, MAX_BUF_LEN - 1, "SET counter:%d:%lld %d", (int)row["business_type"],
            (int64_t)row["business_id"], (int)row["counter"]);

        ScopedReply reply(redis_exec_.Execute("", cmd_buf));

        LOG_DEBUG("reloadRedis() reload " << row["business_type"] << '-' << row["business_id"] 
            << '-' << row["counter"]);
      }
    }
  } catch(const mysqlpp::Exception & e) {
    LOG_ERROR("ReloadRedis() error:" << e.what());
    return false;
  }
  return true;

}

int32_t GenericCounterHandler::Get(const int32_t business_type, const int64_t business_id) {
  const int MAX_BUF_LEN = 64;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';

  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "GET counter:%d:%lld", business_type, business_id);
    // Bulk reply: the value of key, or nil when key does not exist.
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (reply) {
      if (reply->type == REDIS_REPLY_STRING) {
         LOG_DEBUG(cmd_buf << " success, reply=" << reply->str);
         return atoi(reply->str);
      } else {
         LOG_INFO(cmd_buf << " return NIL");
         return -1;
      }
    } else {
      LOG_WARN(cmd_buf << ", null reply, may be connection err.");
      return -1;
    }
  }
  return 0;
}

bool GenericCounterHandler::Add(const int32_t business_type, const int64_t business_id, const int32_t delta) {
  const int MAX_BUF_LEN = 64;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';

  {
    // serialize
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "INCRBY counter:%d:%lld %d", business_type, business_id, delta);

    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (reply) {
      LOG_DEBUG(cmd_buf << " success, reply=" << reply->integer);
    } else {
      LOG_ERROR(cmd_buf << " fail");
    }
  }
  thread_manager_->add(shared_ptr<concurrency::Runnable>(new UpdateCountTask(this, business_type, business_id, delta)));
  return true;

  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return false;
  }

  mysqlpp::Query query(cp->query());

  // redis_adapter_.CommentAdd(item);
  query << "INSERT INTO generic_counter(business_type, business_id, counter) VALUES (" 
        << business_type << ',' << business_id << ',' << delta << ") ON DUPLICATE KEY UPDATE counter=" 
        << "counter + " << delta;

  bool res = false;
  try {
    res = query.exec();
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR("update counter " << business_id << " error:" << e.what());
  }
  LOG_DEBUG("update counter comment " << business_id << " success=" << res);
  return res;
}

}

int main(int argc, char **argv) {
  LOG_INIT("./counter.log", "DEBUG");
  int port = 9094;

  LOG_INFO("listens on port " << port);
  using namespace jztzp;
  ThriftService<GenericCounterHandler, GenericCounterProcessor> service;
  service.Start(port);
  return 0;
}


/*
extern "C" {

void ServiceEntry() {
  int port = 9094;

  using namespace jztzp;
  ThriftService<GenericCounterHandler, GenericCounterProcessor> service;
  service.Start(port);
}

}
*/
